package com.tn.openrpc.jms;

import java.util.concurrent.ExecutorService;
import javax.jms.Message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An implementation of <code>MessageProcessor</code> that performs the message processing on a separate thread to the one that invoked it.
 */
public class ParallelMessageProcessor<M extends Message> implements MessageProcessor<M>
{
  private static final Logger LOGGER = LoggerFactory.getLogger(ParallelMessageProcessor.class);

  private ExecutorService executorService;
  private MessageProcessor<M> delegateMessageProcessor;

  /**
   * Creates a new <code>ParallelMessageProcessor</code>.
   *
   * @param executorService          the executor service that manages the threads used to process the messages,
   * @param delegateMessageProcessor the delegate message processor that actually does the processing.
   */
  public ParallelMessageProcessor(ExecutorService executorService, MessageProcessor<M> delegateMessageProcessor)
  {
    this.executorService = executorService;
    this.delegateMessageProcessor = delegateMessageProcessor;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void processMessage(M message) throws MessageProcessingException
  {
    executorService.submit(new MessageProcessingRunnable(message));
  }

  /**
   * An implementation of <code>Runnable</code> that invokes the delegate message processor.
   */
  private class MessageProcessingRunnable implements Runnable
  {

    private M message;

    /**
     * Creates a new <code>MessageProcessingRunnable</code> initialized with the <code>message</code> to process.
     */
    public MessageProcessingRunnable(M message)
    {
      this.message = message;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void run()
    {
      try
      {
        delegateMessageProcessor.processMessage(message);
      }
      catch (MessageProcessingException e)
      {
        LOGGER.error("An error occurred processing the message.", e);
      }
    }
  }
}
